
package kafka;

import static org.springframework.boot.test.context.SpringBootTest.WebEnvironment.RANDOM_PORT;

import com.huaweicloud.saaslanding.basic.utils.SpringUtils;
import com.huaweicloud.saaslanding.install.SaaslandingInstallApplication;
import com.huaweicloud.saaslanding.install.applicationservice.KafkaService;
import com.huaweicloud.saaslanding.install.properties.Lts;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.ConsumerGroupDescription;
import org.apache.kafka.clients.admin.TopicDescription;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Collections;
import java.util.Map;

/**
 * 功能描述
 *
 * @since 2022-12-05
 */
@Slf4j
@RunWith(SpringRunner.class)
@SpringBootTest(classes = SaaslandingInstallApplication.class, webEnvironment = RANDOM_PORT)
public class KafkaTest {
    @Resource
    private KafkaService kafkaService;

    @Resource
    private Lts lts;

    @Test
    public void deleteKafka() {
        if (lts.isEnabled()) {
            log.info("log enabled, delete kafka...");
            kafkaService.deleteKafka();
        } else {
            log.info("log disabled, skip deleting.");
        }
    }

    @Test
    public void describeKafka() throws Exception {
        if (!lts.isEnabled()) {
            return;
        }

        AdminClient adminClient = SpringUtils.getBean(AdminClient.class);

        Map<String, TopicDescription> topicDescriptionMap =
            adminClient.describeTopics(Collections.singleton("bugfix-cse-keycloak-0104-002")).allTopicNames().get();
        if (topicDescriptionMap.isEmpty()) {
            return;
        }
        log.info("describeTopic: {}", topicDescriptionMap);

        Integer partitions = topicDescriptionMap.get("bugfix-cse-keycloak-0104-002").partitions().size();
        log.info("partition number: {}", partitions);

        Map<String, ConsumerGroupDescription> consumerDescriptionMap =
            adminClient.describeConsumerGroups(Collections.singleton("Transfer-bugfix-cse-keycloak-0104-002"))
                .all()
                .get();
        log.info("descriptionConsumerGroup: {}", consumerDescriptionMap);

        Integer consumers = consumerDescriptionMap.get("Transfer-bugfix-cse-keycloak-0104-002").members().size();
        log.info("consumer number: {}", consumers);
    }
}
